package com.qkjia.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @Auther 赵乾坤
 * @Date 2021-10-01 18:27
 * kafka api 简单消费者 自动提交
 * 手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。
 * 两者的相同点是，都会将本次poll的一批数据最高的偏移量提交；不同点是，commitSync会失败重试，
 * 一直到提交成功（如果由于不可恢复原因导致，也会提交失败）；而commitAsync则没有失败重试机制，故有可能提交失败。
 *
 *
 * offset失效的情况 ：① 换了一个消费组
 * ②消费的时候已经不存在了 ，比如；默认7天失效
 */
public class MyConsumer {
    public static void main(String[] args) {

        // 1.创建消费者配置信息
        Properties properties = new Properties();

        // 2.给配置信息赋值
        // 指定连接的kafka集群
//        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
//        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "8.135.84.118:9092");
        // 开启自动提交
//        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 关闭自动提交 offset 没手动提交会 重复消费
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 自动提交offset延时  1000毫秒延迟，1秒钟的延迟
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

        // key,value的反序列化
//        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG
//                , "org.apache.kafka.common.serialization.StringSerializer");
//        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
//                "org.apache.kafka.common.serialization.StringSerializer");


        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");


        // 加了不爆错
//        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
//        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");

        // 消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "bigdata");

        // 重置消费组的offset
//        offset重置 ①换消费组组 ②重置 已经不存在了
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 创建一个消费者对象，用来消费数据
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        // 订阅主题  Collections.singletonList 数组只有一个值的写法
        //   consumer.subscribe(Collections.singletonList("first"));
        consumer.subscribe(Arrays.asList("first", "second"));

        // 获取数据 :一次拉取会是多个数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(100);

            // 解析并打印ConsumerRecords
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {

                System.out.println(consumerRecord.key() + "--" + consumerRecord.value());

            }
        }

        // 死循环拉取 不用关闭连接
//        consumer.close();

    }
}
